/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017-2027 Chinazyjr,Inc.All Rights Reserved.
 */
package com.sys.midware.MQ.rocketMQ.factory;

import com.sys.midware.MQ.rocketMQ.config.ConfigRocketUtil;
import com.sys.midware.MQ.rocketMQ.logic.IPushConsumer;
import com.sys.midware.MQ.rocketMQ.logic.impl.PushConsumerImpl;
import com.sys.midware.assist.utils.string.StringUtil;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 
 * <b>ClassName：</b> ConsumerFactory <br/>
 * <b>Description：</b> 获取默认的消息消费实例 <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午2:55:46 <br/>
 * <b>@version: </b>  <br/>
 */
public class PushConsumerFactory {
    private static final Logger logger = LoggerFactory.getLogger(PushConsumerFactory.class);

	/**
	 * IConsumer 接口定义
	 */
	private static IPushConsumer consumer = null;

	private PushConsumerFactory(){}
	
	/**
	 * 获取消息消费实例
	 * 
	 * @param messageListener
	 *            具体消费实现着
	 * @param topics
	 *            消费的topic
	 * @return IConsumer
	 */
	public static IPushConsumer getConsumerInstace() {
		return getConsumerInstace(null);
	}

	/**
     * 根据分组和实例获取实例 
     * @param  
     * @return  IPushConsumer
     */
    public static IPushConsumer getConsumerInstace(String instance,String consumerGroup) {

        return getConsumerInstace(instance,consumerGroup,null);
    }
    
    public static IPushConsumer getConsumerInstace(String instance,String consumerGroup,String clientIp){
        if (null != consumer) {
            return consumer;
        }

        if (StringUtil.isEmpty(consumerGroup)) {
            consumerGroup = ConfigRocketUtil.getMQGroupConsumer();
        }
        if (StringUtil.isEmpty(instance)) {
            instance = ConfigRocketUtil.getMQGroupConsumer();
        }
        
        synchronized (PushConsumerFactory.class) {
            if (null == consumer) {
             // set name server
                String  nameServer = ConfigRocketUtil.getRocketNameServer();
                consumer = new PushConsumerImpl(nameServer,instance,consumerGroup,clientIp);
            }
            Runtime.getRuntime().addShutdownHook(new CleanHook());
        }
        return consumer;
    }
	
	/**
	 * 根据分组获取实例 
	 * @param  
	 * @return  IPushConsumer
	 */
	public static IPushConsumer getConsumerInstace(String consumerGroup) {
        return getConsumerInstace(null,consumerGroup);
    }
	
	static void clean() {
        try {
            logger.info("开始关闭PushConsumer...");
            consumer.shutdown();
            logger.info("结束关闭PushConsumer...");
        } catch (Exception e) {
            logger.error("关闭PushConsumer异常", e);
        }
    }

    static class CleanHook extends Thread {
        public void run() {
            clean();
        }
    }
}
